SNS＋SQS＋DynamoDBでハードバウンスと苦情の発生を記録する環境をCDK(TypeScript)で作成してみる

#AWS

#Amazon SES

カトアキ

2020.07.17

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

はじめに

awsでSESを使う場合には、ハードバウンスに対応する必要があります。対応のためには発生の検知と、検知したメールアドレスに対して繰り返し送信を行わないようにすることの両方が必要となりますが、今回は発生を検知して記録する仕組みまでを構築したいと思います。

awsのブログ記事¹を参考に、少し変更を加えた形で構築します。

前提

SESで利用するドメイン設定が行われていること
- 本記事では katoaki.com が検証済みであるとします
TypeScriptを利用します
CDK v1.38.0を利用します
- 各種モジュールもすべてv1.38.0に合わせます
SESはオレゴンリージョンを使用します

やってみたこと

全体設計

各サービスを以下の順番で連携します
- SES → SNS → SQS → Lambda → DynamoDB
各サービスの役割は次の通り
- SES
  - バウンス・苦情が発生した場合に、バウンス・苦情用SNSトピックをトリガします
- SNS
  - SESからメッセージが発行された場合に、SQSのバウンス・苦情キューにメッセージを送信します
- SQS
  - SNSからメッセージが発行された場合に、バウンス・苦情のDB保存用Lambdaをトリガします
- Lambda
  - SQSからトリガされた場合に、SESの通知用テーブルへバウンス・苦情の情報を登録します
- DynamoDB
  - 送信先メールアドレスと発生時刻を複合キーとして、Lambdaから登録されたバウンス・苦情の情報を保持します

CDK init & 必要モジュールのインストール

bash command

cdk init --language typescript

bash command

npm i "@aws-cdk/aws-sqs@1.38.0"
npm i "@aws-cdk/aws-sns@1.38.0"
npm i "@aws-cdk/aws-iam@1.38.0"
npm i "@aws-cdk/aws-dynamodb@1.38.0"
npm i "@aws-cdk/aws-lambda@1.38.0"

cdk-stack.tsのスケルトンに必要なモジュールのimport文を追加し、リソース作成用の関数 create() を追加して以下のようにします。

cdk-stack.ts

import * as cdk from '@aws-cdk/core';
import sqs = require("@aws-cdk/aws-sqs");
import sns = require("@aws-cdk/aws-sns");
import iam = require("@aws-cdk/aws-iam");
import ddb = require("@aws-cdk/aws-dynamodb");
import lambda = require("@aws-cdk/aws-lambda");

export class App1Stack extends cdk.Stack {
  constructor(scope: cdk.Construct, id: string, props?: cdk.StackProps) {
    super(scope, id, props);

    // The code that defines your stack goes here
    this.create();
  }
  private create(){
    
  }
}

SQSのキューを作成

キューのリソースを定義します。その上で、SNSトピックからメッセージが送信できるようにするには、キューポリシーの設定が必須なので定義します。

create() 関数内に以下を追加してください。

cdk-stack.ts

// SQS
const BounceQueue = new sqs.Queue(this, "BounceQueue", {
  queueName: "BounceQueue"
});
const BounceQueuePolicy = new iam.PolicyStatement();
BounceQueuePolicy.addActions("SQS:SendMessage");
BounceQueuePolicy.addResources(BounceQueue.queueArn);
BounceQueuePolicy.addPrincipals(new iam.AnyPrincipal());
BounceQueue.addToResourcePolicy(BounceQueuePolicy);
const ComplaintQueue = new sqs.Queue(this, "ComplaintQueue", {
  queueName: "ComplaintQueue"
});
const ComplaintQueuePolicy = new iam.PolicyStatement();
ComplaintQueuePolicy.addActions("SQS:SendMessage");
ComplaintQueuePolicy.addResources(ComplaintQueue.queueArn);
ComplaintQueuePolicy.addPrincipals(new iam.AnyPrincipal());
ComplaintQueue.addToResourcePolicy(ComplaintQueuePolicy);

SNSのトピックを作成

SNSトピックを定義します。その上で、トピックにメッセージが発行された際にSQSをトリガする（メッセージを送信する）ようにするために、サブスクリプションにキューを設定します。

create() 関数内に以下を追加してください。

cdk-stack.ts

const BounceTopic = new sns.Topic(this, "BounceTopic", {
  topicName: "BounceTopic"
})
new sns.Subscription(this, "BounceSubscription", {
  endpoint: BounceQueue.queueArn,
  protocol: sns.SubscriptionProtocol.SQS,
  topic: BounceTopic
})
const ComplaintTopic = new sns.Topic(this, "ComplaintTopic", {
  topicName: "ComplaintTopic"
})
new sns.Subscription(this, "ComplaintSubscription", {
  endpoint: ComplaintQueue.queueArn,
  protocol: sns.SubscriptionProtocol.SQS,
  topic: ComplaintTopic
})

Lambda関数を作成

キューから受け取った内容を処理してテーブルへ書き込むため、Lambda関数を作成します。

ポイントは以下です

SQSからトリガされるように、EventSourceMappingの設定を行う
SQSからのメッセージを読めるようにポリシーの設定を行う
DynamoDBへ書き込みを行えるようにポリシーの設定を行う
処理を別ファイルに分けてcodeで指定する
cdk-stack.ts

// Lambda
const ReportLambdaPolicy = new iam.PolicyStatement();
ReportLambdaPolicy.addActions(
  "sqs:ReceiveMessage",
  "dynamodb:PutItem"
);
ReportLambdaPolicy.addResources(
  `arn:aws:dynamodb:*:*:table/ses_notifications`,
  BounceQueue.queueArn,
  ComplaintQueue.queueArn     
);
const ReportLambda = new lambda.Function(this, "ReportLambda", {
  functionName: "ReportLambda",
  timeout: cdk.Duration.seconds(10),
  code: lambda.Code.asset("src/index.zip"),
  handler: "index.handler",
  runtime: lambda.Runtime.NODEJS_12_X,
  initialPolicy: [ReportLambdaPolicy],
})
new lambda.EventSourceMapping(this, "BounceEventSourceMapping", {
  batchSize: 10,
  enabled: true,
  eventSourceArn: BounceQueue.queueArn,
  target: ReportLambda
})
new lambda.EventSourceMapping(this, "ComplaintEventSourceMapping", {
  batchSize: 10,
  enabled: true,
  eventSourceArn: ComplaintQueue.queueArn,
  target: ReportLambda
})

Lambda関数の実装部分です。こちらは同じディレクトリにzip化しておく必要があります。

ポイントは以下です

AWSのブログ記事¹を参考に作成しました
当該記事はSNSから直接Lambdaをトリガする構成なので、今回の用途に合わせて修正が必要でした
exports.handler〜の部分配下の5行を修正しています

修正部分：

exports.handler = function (event, context) {
  console.log("Received event:", JSON.stringify(event, null, 2));
  var body = JSON.parse(event.Records[0].body);
  var SnsPublishTime = body.Timestamp;
  var SnsTopicArn = body.TopicArn;
  var SESMessage = body.Message;

src/index.js

console.log("Loading event");
var aws = require("aws-sdk");
var ddb = new aws.DynamoDB({ params: { TableName: "ses_notifications" } });
exports.handler = function (event, context) {
  console.log("Received event:", JSON.stringify(event, null, 2));
  var body = JSON.parse(event.Records[0].body);
  var SnsPublishTime = body.Timestamp;
  var SnsTopicArn = body.TopicArn;
  var SESMessage = body.Message;
  SESMessage = JSON.parse(SESMessage);
  var SESMessageType = SESMessage.notificationType;
  var SESMessageId = SESMessage.mail.messageId;
  var SESDestinationAddress = SESMessage.mail.destination.toString();
  var LambdaReceiveTime = new Date().toString();
  if (SESMessageType == "Bounce") {
    var SESreportingMTA = SESMessage.bounce.reportingMTA;
    var SESbounceSummary = JSON.stringify(SESMessage.bounce.bouncedRecipients);
    var itemParams = {
      Item: {
        SESMessageId: { S: SESMessageId },
        SnsPublishTime: { S: SnsPublishTime },
        SESreportingMTA: { S: SESreportingMTA },
        SESDestinationAddress: { S: SESDestinationAddress },
        SESbounceSummary: { S: SESbounceSummary },
        SESMessageType: { S: SESMessageType },
      },
    };
    ddb.putItem(itemParams, function (err, data) {
      if (err) {
        context.fail(err);
      } else {
        console.log(data);
        context.succeed();
      }
    });
  } else if (SESMessageType == "Delivery") {
    var SESsmtpResponse1 = SESMessage.delivery.smtpResponse;
    var SESreportingMTA1 = SESMessage.delivery.reportingMTA;
    var itemParamsdel = {
      Item: {
        SESMessageId: { S: SESMessageId },
        SnsPublishTime: { S: SnsPublishTime },
        SESsmtpResponse: { S: SESsmtpResponse1 },
        SESreportingMTA: { S: SESreportingMTA1 },
        SESDestinationAddress: { S: SESDestinationAddress },
        SESMessageType: { S: SESMessageType },
      },
    };
    ddb.putItem(itemParamsdel, function (err, data) {
      if (err) {
        context.fail(err);
      } else {
        console.log(data);
        context.succeed();
      }
    });
  } else if (SESMessageType == "Complaint") {
    var SESComplaintFeedbackType = SESMessage.complaint.complaintFeedbackType;
    var SESFeedbackId = SESMessage.complaint.feedbackId;
    var itemParamscomp = {
      Item: {
        SESMessageId: { S: SESMessageId },
        SnsPublishTime: { S: SnsPublishTime },
        SESComplaintFeedbackType: { S: SESComplaintFeedbackType },
        SESFeedbackId: { S: SESFeedbackId },
        SESDestinationAddress: { S: SESDestinationAddress },
        SESMessageType: { S: SESMessageType },
      },
    };
    ddb.putItem(itemParamscomp, function (err, data) {
      if (err) {
        context.fail(err);
      } else {
        console.log(data);
        context.succeed();
      }
    });
  }
};

DynamoDBテーブルを作成

create() 関数内に以下を追加します。

//DynamoDB
new ddb.Table(this, "SesNotificationsTable", {
  tableName: "ses_notifications",
  partitionKey: {
    name: "SESMessageId",
    type: ddb.AttributeType.STRING,
  },
  sortKey: {
    name: "SnsPublishTime",
    type: ddb.AttributeType.STRING,
  },
});

SESの設定

マネジメントコンソールからNotificationsに、BounceTopic・ComplaintTopicを指定します
- この設定は、執筆時点ではCDKで定義する方法が見つからなかったため手動で行っています

binの設定

cdk deploy コマンドで作成するスタックの設定で、オレゴンリージョンに作成するように設定変更します。SESのリージョンに合わせるためです。

new App1Stack(app, 'App1Stack', {
    env: {
      region: "us-west-2",
    }
  }
)

おわりに

少し手間ではありますが、一度作成すれば他のAWS環境にも流用できるので一度作っておくと便利だと思います。

参考

[AWS Lambda 関数を使用してSES を DynamoDB に保存する] (https://aws.amazon.com/jp/premiumsupport/knowledge-center/lambda-sns-ses-dynamodb/) ↩ ↩

SNS＋SQS＋DynamoDBでハードバウンスと苦情の発生を記録する環境をCDK(TypeScript)で作成してみる

はじめに

前提

やってみたこと

全体設計

CDK init & 必要モジュールのインストール

SQSのキューを作成

SNSのトピックを作成

Lambda関数を作成

DynamoDBテーブルを作成

SESの設定

binの設定

おわりに

参考

関連記事

主なカテゴリ

AWS

おすすめ

プロダクト

コンテンツ

お問い合わせ

運営会社